package net.bwie.realtime.jtp.dwd.logs.car.fuction;

import net.bwie.realtime.jtp.dwd.logs.car.job.CarBean;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class CarViewWindowFunction implements WindowFunction<CarBean, String, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<CarBean> iterable, Collector<String> collector) throws Exception {
        //窗口开始时间
        String windowStart = DateTimeUtil.convertLongToString(timeWindow.getStart(), "yyyy-MM-dd HH:mm:ss");
        //窗口结束时间
        String windowEnd = DateTimeUtil.convertLongToString(timeWindow.getEnd(), "yyyy-MM-dd HH:mm:ss");
        //输出
        CarBean bean = iterable.iterator().next();
        String vin = bean.getVin();
        Long timestamp = bean.getTimestamp();
        int type = bean.getType();
        double longitude = bean.getLongitude();
        double latitude = bean.getLatitude();
        double speed = bean.getSpeed();
        double mileage = bean.getMileage();
        double batteryLevel = bean.getBatteryLevel();
        double batteryTemp = bean.getBatteryTemp();
        double motorTemp = bean.getMotorTemp();
        String chargingStatus = bean.getChargingStatus();
        double energyConsumption = bean.getEnergyConsumption();
        String output = windowStart + "," + windowEnd + "," + vin + "," + timestamp + "," +
                type + "," + longitude + "," + latitude + "," + speed + "," + mileage + "," +
                batteryLevel + "," + batteryTemp + "," + motorTemp + "," + chargingStatus
                + "," + energyConsumption;
        collector.collect(output);
    }
}

